package com.tree.redisson.runner;

import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.tree.redisson.IUserService;
import com.tree.redisson.entity.User;
import lombok.extern.slf4j.Slf4j;
import org.redisson.RedissonShutdownException;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Description: load plan 队列线程（处理等待状态的数据）  MasterSchedulerService
 * <br>
 * <p>
 * 生产者 、消费者
 *
 * @Author frankzsliu
 * @Date 2020/11/3 5:33 下午
 * @Version 1.0
 */
@Slf4j
@Component
public class LoadPlanConsumerThread extends Thread {

    private static final String LOAD_PLAN_LOCK = "LOCK:LOADPLAN";

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 核心线程数
     */
    private static final int LOAD_PLAN_THREAD_SIZE = 10;

    /**
     * master exec service
     */
    private ExecutorService loadPlanExecutor = Executors.newFixedThreadPool(
            LOAD_PLAN_THREAD_SIZE,
            new ThreadFactoryBuilder().setDaemon(true)
                    .setNameFormat("LoadPlan-Exec-Thread")
                    .build());

    private List<User> userList;

    @Autowired
    private IUserService userService;


    @Override
    public void run() {
        //TODO 设置停止标识
        log.info("LoadPlanQueueThread started ...");
        Random rand = new Random();
        while (true) {
            RLock lock = redissonClient.getLock(LOAD_PLAN_LOCK);
            boolean flag = false;
            try {
                flag = lock.tryLock(3, TimeUnit.SECONDS);
                if (flag) {
                    QueryWrapper queryWrapper = new QueryWrapper();
                    queryWrapper.eq("status", -1);
                    userList = userService.list(queryWrapper);
                    List<Integer> ids = userList.stream().map(user->user.getId()).collect(Collectors.toList());
                    log.info("获取锁成功，开始消费用户【" + Thread.currentThread() + "】, 数量【" + userList.size() + "】,需要消费的用户详情为" + JSONUtil.toJsonPrettyStr(ids));
                    //遍历丢入线程池执行
                    userList.forEach(user -> {
                        loadPlanExecutor.submit(() -> {
                            //模拟处理业务逻辑处理时间
                            int sleepTime = rand.nextInt(10);
                            try {
                                Thread.sleep(sleepTime * 1000);
                                user.setStatus(0);
                                userService.updateById(user);
                                log.info("完成单个用户消费,需要消费的用户详情为" + JSONUtil.toJsonPrettyStr(user) + "， 休息【" + sleepTime + "】秒");
                            } catch (InterruptedException interruptedException) {
                                interruptedException.printStackTrace();
                            }
                        });
                    });
                } else {
                    log.warn("locked[false], task already locked by other node， skip over...");
                }

//                int sleepTime = rand.nextInt(10);
//                //indicate that no command ,sleep for 1s
//                log.info("finish current forloop, sleep[" + sleepTime + "]s... ");
//                Thread.sleep(sleepTime * 1000);

            } catch (Throwable e) {
                log.warn("task supervisor threw an exception", e);
            } finally {
                //只有当前线程获取了才能解除分布式锁
                if (flag) {
                    lock.unlock();
                    log.debug("end execute SchedulerSupervisorTask run method and unlock ....");
                }

            }
        }


    }

}
